package com.example.wxg.example.stock;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.*;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * @author void
 * @date 2022/1/5 15:19
 * @desc
 */
public class StockSource implements SourceFunction<StockPrice> {
    
    //Source是否正在执行
    private boolean isRunning = true;
    
    //数据集文件名
    private String path;
    private InputStream streamSource;

    public StockSource(String path) {
        this.path = path;
    }

    /**
     * 读取数据集中的元素，每隔一定时间发送一次股票数据
     * @param sourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceContext<StockPrice> sourceContext) throws Exception {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd HHmmss");
        //从项目的resources目录获取输入
//        streamSource = this.getClass().getClassLoader().getResourceAsStream(path);
        streamSource = new FileInputStream(path);
        BufferedReader br = new BufferedReader(new InputStreamReader(streamSource));
        
        String line;
        boolean isFirstLine = true;
        long timeDiff = 0L;
        long lastEventTs = 0L;
        while (isRunning && (line=br.readLine())!=null){
            String[] itemStrArr = line.split(",");
            LocalDateTime dateTime = LocalDateTime.parse(itemStrArr[1]+" "+itemStrArr[2], formatter);
            long eventTs = Timestamp.valueOf(dateTime).getTime();
            if(isFirstLine){
                //从第一行数据提取时间戳
                lastEventTs = eventTs;
                isFirstLine = false;
            }
            
            StockPrice stock = new StockPrice();
            stock.setSymbol(itemStrArr[0]);
            stock.setTs(eventTs);
            stock.setPrice(Double.parseDouble(itemStrArr[3]));
            stock.setVolume(Integer.parseInt(itemStrArr[4]));
            //输入文件中的时间戳是从小到大排列的
            //新读取的行如果比上一行大，则等待，以此模拟一个有时间间隔的输入流
            timeDiff = eventTs - lastEventTs;
            if (timeDiff > 0) {
                Thread.sleep(timeDiff);
            }
            sourceContext.collect(stock);
            lastEventTs = eventTs;
        }
        System.out.println(sourceContext);
    }

    /**
     * 停止发送数据
     */
    @Override
    public void cancel() {
        try {
            streamSource.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        isRunning = false;
    }
}
